package com.example.java.sink;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class MysqlSink extends RichSinkFunction<Student> {
    private DruidPooledConnection connection =null;
    private PreparedStatement preparedStatement =null;
    private DruidDataSource dataSource = null;
    // 与数据源建立连接
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = getConnection();
        String sql = "insert into student values(?,?,?)";
        preparedStatement = connection.prepareStatement(sql);
        super.open(parameters);
    }

    private DruidPooledConnection getConnection() throws SQLException {
        dataSource = new DruidDataSource();
        dataSource.setUrl("jdbc:mysql://linux123:3306/test?useSSL=false");
        dataSource.setUsername("root");
        dataSource.setPassword("12345678");

        //下面都是可选的配置
        dataSource.setInitialSize(10);  //初始连接数，默认0
        dataSource.setMaxActive(30);  //最大连接数，默认8
        dataSource.setMinIdle(10);  //最小闲置数
        dataSource.setMaxWait(2000);  //获取连接的最大等待时间，单位毫秒
        dataSource.setPoolPreparedStatements(true); //缓存PreparedStatement，默认false
        //缓存PreparedStatement的最大数量，默认-1（不缓存）。
        // 大于0时会自动开启缓存PreparedStatement，所以可以省略上一句代码
        dataSource.setMaxOpenPreparedStatements(20);
        //获取连接
        return dataSource.getConnection();
    }

    // 每一条记录会被调用一次
    @Override
    public void invoke(Student value, Context context) throws Exception {
        preparedStatement.setInt(1, value.getId());
        preparedStatement.setString(2, value.getName());
        preparedStatement.setInt(3, value.getAge());
        preparedStatement.executeUpdate();
    }

    // 关闭资源
    @Override
    public void close() throws Exception {
        super.close();
        if(connection != null) {
            connection.close();
        }
        if(preparedStatement != null) {
            preparedStatement.close();
        }
    }
}
